/*
 * This file is part of the xOperator SPARQL/XMPP agent.
 * For further information see: http://xoperator.aksw.org
 * Copyright (C) 2007-2008  Jörg Unbehauen

 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.

 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.

 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.aksw.xoperator.sparql.local;

import com.hp.hpl.jena.query.DataSource;
import com.hp.hpl.jena.query.Dataset;
import com.hp.hpl.jena.query.DatasetFactory;
import com.hp.hpl.jena.query.Query;
import com.hp.hpl.jena.query.QueryExecution;
import com.hp.hpl.jena.query.QueryExecutionFactory;
import com.hp.hpl.jena.query.QueryFactory;
import com.hp.hpl.jena.query.ResultSet;
import com.hp.hpl.jena.query.ResultSetFormatter;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.util.FileManager;
import com.hp.hpl.jena.util.LocatorFile;

import org.aksw.xoperator.conf.AgentConfiguration;
import org.aksw.xoperator.roster.IRosterManager;
import org.aksw.xoperator.sparql.AbstractThreadPooledSparqler;
import org.aksw.xoperator.sparql.SparqlEndpointIdentifier;
import org.aksw.xoperator.sparql.SparqlEndpointType;
import org.aksw.xoperator.sparql.SparqlExecutionFuture;
import org.aksw.xoperator.sparql.SparqlQuery;
import org.aksw.xoperator.sparql.SparqlResult;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.jivesoftware.smack.SmackConfiguration;

import groovy.sql.DataSet;

import java.io.File;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;


/**
 * The local store using a jena/arq for answering queries. Loads data from a folder in the users conf directory. In this store the a triple representation of the roster is held.
 *
  */
public class LocalSparqlEndpoint extends AbstractThreadPooledSparqler {
    public static SparqlEndpointIdentifier id = new SparqlEndpointIdentifier(SparqlEndpointType.Local,
            "local");
    private AgentConfiguration conf;
    private IRosterManager rosterManager;
    private Model defaultModel = null;
    private Log log = LogFactory.getLog(LocalSparqlEndpoint.class);
    private FileManager filemanager;
    
    private int timeout = SmackConfiguration.getPacketReplyTimeout();
    
    
    public void setTimeout(int timeout) {
		this.timeout = timeout;
	}
    
    public int getTimeout() {
		return timeout;
	}

    public LocalSparqlEndpoint(AgentConfiguration conf,
        IRosterManager rosterManager, ThreadPoolExecutor executor) {
        super(executor);
        this.conf = conf;
        this.rosterManager = rosterManager;

        filemanager = new FileManager();
        filemanager.setModelCaching(true);
        createDataSet();
    }

    private void createDataSet() {
        log.info("Creating the dataset");

        File rdfFolder = new File(conf.getConfDir() + "/rdf/default");
        
        //these two components are working on the same model
        defaultModel = rosterManager.rosterAsModel();

        filemanager.addLocator(new LocatorFile(rdfFolder.getAbsolutePath()));

        for (File rdfFile : Arrays.asList(rdfFolder.listFiles())) {
            if (!rdfFile.isDirectory()) {
            	defaultModel.add(filemanager.loadModel(rdfFile.getName()));
            }
        }

        for (String trustedDocLoc : this.conf.getLocalSparqlEndpoints()
                                             .getTrustedDocuments()) {
        	defaultModel.add(filemanager.loadModel(trustedDocLoc));
        }

      
        log.info("Finshed creating the dataset");
    }


    /* (non-Javadoc)
     * @see org.aksw.xoperator.sparql.SparqlEndpoint#getFuturesFor(org.aksw.xoperator.sparql.SparqlQuery)
     */
    public List<SparqlExecutionFuture<SparqlResult>> getFuturesFor(SparqlQuery query) {
        List<SparqlExecutionFuture<SparqlResult>> queries = new ArrayList<SparqlExecutionFuture<SparqlResult>>();

        if (query.isAskLocal()) {
            queries.add(new SparqlExecutionFuture<SparqlResult>(new LocalCallable(query),"local",query.getQueryString()));
        }

        return queries;
    }


    /* (non-Javadoc)
     * @see org.aksw.xoperator.sparql.SparqlEndpoint#process(org.aksw.xoperator.sparql.SparqlQuery)
     */
    public void process(SparqlQuery sparqlQuery) {
        this.processFutures(this.getFuturesFor(sparqlQuery), timeout
            );
    }

    private class LocalCallable implements Callable<SparqlResult> {
        SparqlQuery sparqlQuery;

        public LocalCallable(SparqlQuery sparqlQuery) {
            this.sparqlQuery = sparqlQuery;
        }

        public SparqlResult call() throws Exception {
            log.info("Querying local endpoint");

            Query query = QueryFactory.create(this.sparqlQuery.getQueryString());
            Dataset data;
            if(query.getBaseURI()!=null){
            	data = DatasetFactory.create(query.getBaseURI(),query.getNamedGraphURIs());
            	
            }else{
            	
            	            	
            	DataSource ds = DatasetFactory.create(defaultModel);
            	 for (Object uriObj : query.getNamedGraphURIs()) {
                 	ds.addNamedModel((String) uriObj,filemanager.loadModel((String) uriObj));
     			}
            	 data = ds;
            }
             
       
            QueryExecution quexec = QueryExecutionFactory.create(query, data);
            SparqlResult sparqlResult = new SparqlResult();
            sparqlResult.setAnsweredBy(id);
            sparqlQuery.addResult(sparqlResult);

            try {
                ResultSet resultSet = quexec.execSelect();

                String result = ResultSetFormatter.asXMLString(resultSet);
                sparqlResult.setAnswer(result);
            } catch (RuntimeException e) {
                log.error("Error querying the local store: ", e);
                sparqlResult.setErrorMsg("Error querying the local store: " +
                    e.getMessage());
            }

            log.info("Finished Quering the local endpoint");

            return sparqlResult;
        }
    }
}
